package org.funtester.performance.books.chapter09.section3;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.funtester.performance.books.chapter03.section3.ThreadTask;
import org.funtester.performance.books.chapter03.section4.TaskExecutor;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

/**
 * 消费者性能测试用例
 */
public class ConsumerCase {

    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();// 配置信息
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");// 指定kafka集群地址
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "FunTester");// 指定消费者组
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// key的序列化器
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// value的序列化器
        properties.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");// 自动提交的时间间隔
        properties.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "10");// 每次最大拉取的条数
        properties.setProperty(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "10000");// 消费者会话超时时间
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");// 消费者组不存在时，从最早的消息开始消费
        String topic = "user_behavior";// 指定topic
        int totalNum = 1000;// 总执行次数
        int threadNum = 8;// 线程数
        int rumpUpTime = 20;// Rump-Up运行时间
        int max_wait = 1000;// 拉取数据的最大等待时间,单位毫秒
        List<ThreadTask> tasks = new ArrayList<>();// 任务集合
        ThreadLocal<KafkaConsumer<String, String>> consumerThreadLocal = ThreadLocal.withInitial(() -> {// 创建消费者对象threadLocal对象,并且设置初始化方法
            KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);// 创建消费者对象
            kafkaConsumer.subscribe(Arrays.asList(topic));// 消费者订阅主题
            return kafkaConsumer;
        });// 创建线程本地变量
        for (int i = 0; i < threadNum; i++) {
            ThreadTask threadTask = new ThreadTask() {

                @Override
                public void test() {
                    ConsumerRecords<String, String> records = consumerThreadLocal.get().poll(Duration.ofMillis(max_wait));// 拉取数据,超时设置,拉取不到数据则返回空
                    for (ConsumerRecord<String, String> record : records) {// 遍历拉取到的数据
                        UserBehavior.parse(record.value());// 解析消息
                        // 处理消息逻辑,此处省略
                    }

                }

            };
            threadTask.totalNum = totalNum;// 设置执行次数
            threadTask.costTime = new ArrayList<>(totalNum);// 设置任务的执行时间集合,设置容量,避免频繁扩容
            tasks.add(threadTask);// 添加任务到任务集合
        }
        TaskExecutor taskExecutor = new TaskExecutor(tasks, "用户行为买点信息消费者", rumpUpTime);
        taskExecutor.start();// 启动任务执行器
    }


}
